﻿using Confluent.Kafka;
using LJ.Abstractions.Config;
using LJ.StreamSystem.Data;
using LJ.StreamSystem.LazyHare.Transport.Abstractions;
using Streamiz.Kafka.Net;
using Streamiz.Kafka.Net.SerDes;
using Streamiz.Kafka.Net.Stream;

namespace LJ.StreamSystem.LazyHare.Transport.Kafka.Internal;

public class KafkaStreamSystemClient(IConfigProvider configProvider) : IStreamSystemClient
{
    private string _applicationId = null!;
    private string _aggregateSummaryTopic = null!;
    private string[] _aggregateTopics = null!;

    private bool disposedValue;

    private KafkaStream? _kafkaStream;

    private readonly StreamSystemTransportConfig _streamSystemTransportConfig = configProvider.Get<StreamSystemTransportConfig>();

    public Func<StreamSystemTransportData, Task<string>>? OnKeyCallback { get; set; }

    public Func<StreamSystemTransportData,Task<string>>? OnInitDataCallback { get; set; }

    public Func<StreamSystemTransportData, Task<string>>? OnSummaryDataCallback { get; set; }


    internal void Init(string applicationId, string aggregateSummaryTopic, string[] aggregateTopics)
    {
        _applicationId = applicationId;
        _aggregateSummaryTopic = aggregateSummaryTopic;
        _aggregateTopics = aggregateTopics;
    }

    public async Task ListeningAsync(CancellationToken cancellationToken)
    {
        cancellationToken.ThrowIfCancellationRequested();

        var config = Config();

        var builder = new StreamBuilder();
        IKStream<string, string>? stream = FetchTopic(builder);

        ArgumentNullException.ThrowIfNull(stream, nameof(stream));

        stream.Map((k, v, _) =>
        {
            return KeyValuePair.Create(GetKey(v), v);
        })
        .GroupByKey()
        .Aggregate<string, StringSerDes>(
            () =>
            {
                return GetInitData();
            },
            (k, v, s) =>
            {
                return GetSummaryData(k, v, s);
            }
        )
        .ToStream()
        .Map((k, v, c) =>
        {
            return KeyValuePair.Create(k, v);
        })
        .To<StringSerDes, StringSerDes>(_aggregateSummaryTopic);

        var topology = builder.Build();

        _kafkaStream = new KafkaStream(topology, config);

        await _kafkaStream.StartAsync();
    }

   

    private string GetInitData()
    {
        var headers = new Dictionary<string, string?>();
        headers[StreamSystemHeaders.ApplicationId] = _applicationId;
        headers[StreamSystemHeaders.AggregateSummaryTopic] = _aggregateSummaryTopic;
        headers[StreamSystemHeaders.AggregateTopics] = string.Join(",", _aggregateTopics);

        var data = new StreamSystemTransportData(headers, string.Empty);

        return OnInitDataCallback!(data).GetAwaiter().GetResult();
    }

    private string GetSummaryData(string k, string v, string s)
    {
        var headers = new Dictionary<string, string?>();
        headers[StreamSystemHeaders.ApplicationId] = _applicationId;
        headers[StreamSystemHeaders.AggregateSummaryTopic] = _aggregateSummaryTopic;
        headers[StreamSystemHeaders.AggregateTopics] = string.Join(",", _aggregateTopics);
        headers[StreamSystemHeaders.Key] = k;

        var data = new StreamSystemTransportData(headers, v);
        data.SummaryBody = s;

        return OnSummaryDataCallback!(data).GetAwaiter().GetResult();
    }

    private string GetKey(string v)
    {
        var headers = new Dictionary<string, string?>();
        headers[StreamSystemHeaders.ApplicationId] = _applicationId;
        headers[StreamSystemHeaders.AggregateSummaryTopic] = _aggregateSummaryTopic;
        headers[StreamSystemHeaders.AggregateTopics] = string.Join(",", _aggregateTopics);

        var data = new StreamSystemTransportData(headers, v);
        return OnKeyCallback!(data).GetAwaiter().GetResult();
    }

    private IKStream<string, string>? FetchTopic(StreamBuilder builder)
    {
        IKStream<string, string>? stream = null;

        var stringSerdes = new StringSerDes();
        foreach (var aggregateTopic in _aggregateTopics)
        {
            if (stream == null)
            {
                stream = builder.Stream(aggregateTopic, stringSerdes, stringSerdes);
            }
            else
            {
                stream = stream.Merge(builder.Stream(aggregateTopic, stringSerdes, stringSerdes));
            }
        }

        return stream;
    }

    private StreamConfig<StringSerDes, StringSerDes> Config()
    {
        var stringSerdes = new StringSerDes();
        var config = new StreamConfig<StringSerDes, StringSerDes>
        {
            ApplicationId = _applicationId,
            BootstrapServers = _streamSystemTransportConfig.BootstrapServers,
            Acks = Acks.All,
            AutoOffsetReset = AutoOffsetReset.Earliest,

            FollowMetadata = true,
            DefaultKeySerDes = stringSerdes,
            DefaultValueSerDes = stringSerdes,
        };

        return config;
    }

    #region Dispose
    protected virtual void Dispose(bool disposing)
    {
        if (!disposedValue)
        {
            if (disposing)
            {
                // TODO: 释放托管状态(托管对象)
                _kafkaStream?.Dispose();
            }

            // TODO: 释放未托管的资源(未托管的对象)并重写终结器
            // TODO: 将大型字段设置为 null
            disposedValue = true;
        }
    }

    // // TODO: 仅当“Dispose(bool disposing)”拥有用于释放未托管资源的代码时才替代终结器
    // ~KafkaEventBusConsumerClient()
    // {
    //     // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
    //     Dispose(disposing: false);
    // }

    public void Dispose()
    {
        // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
        Dispose(disposing: true);
        GC.SuppressFinalize(this);
    }
    #endregion
}
